package bolts;

import java.util.Date;
import java.util.Map;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.tuple.Tuple;

public class OutputUserSimilarityBolt extends AbstractOutputToFileBolt {

    private final String outputPrefix;

    public OutputUserSimilarityBolt(String outputPrefix) {
        super();
        this.outputPrefix = outputPrefix;
    }

    @Override
    protected String getOutputFileName(Map stormConf) {
        return this.outputPrefix + (String) stormConf.get("userSimilarityOutputFile");
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        writer.println(dateFormat.format((Date) input.getValueByField("time")) + ", " + input.getStringByField("user1") + ", "
                + input.getStringByField("user2") + ", " + input.getDoubleByField("score"));
        writer.flush();
    }

}
